Commit f113f221 authored by Vincent Pelletier's avatar Vincent Pelletier

Fix the 2 previously added tests, they were all wrong: test what happen when...

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.


git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@18501 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent 7d9f7f8d
......@@ -40,6 +40,8 @@ from AccessControl.SecurityManagement import newSecurityManager
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,19 +1929,89 @@ class TestCMFActivity(ERP5TypeTestCase):
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, 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, connection_id):
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=['dummy_activity'],
method_id_list=['dummy_activity'],
message_list=['dummy_message'],
path_list=[path],
method_id_list=[method_id],
message_list=[pickled_message],
priority_list=[1],
processing_node_list=[-4],
processing_node_list=[-2],
group_method_id_list=[''],
tag_list=[''],
order_validation_text_list=['']
......@@ -1948,39 +2020,127 @@ class TestCMFActivity(ERP5TypeTestCase):
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_83_ActivityModificationsViaCMFActivityConnectionRolledBackOnErrorSQLDict(self, quiet=0, run=run_all_test):
def test_86_ActivityToolInvokeGroupFailureDoesNotCommitCMFActivitySQLConnectionSQLDict(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.
Check that CMFActivity SQL connection is rollback if activity_tool.invokeGroup raises.
"""
if not run: return
if not quiet:
message = '\nCheck activity modifications via CMFActivity connection are rolled back on error (SQLDict)'
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()
self.activityModificationsViaCMFActivityConnectionRolledBackOnError('SQLDict')
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_84_ActivityModificationsViaCMFActivityConnectionRolledBackOnErrorSQLQeue(self, quiet=0, run=run_all_test):
def test_87_ActivityToolInvokeFailureDoesNotCommitCMFActivitySQLConnectionSQLQueue(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.
Check that CMFActivity SQL connection is rollback if activity_tool.invoke raises.
"""
if not run: return
if not quiet:
message = '\nCheck activity modifications via CMFActivity connection are rolled back on error (SQLQueue)'
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('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()
......
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