Skip to content

erp5
erp5
Commit 0cf9f566 authored by Julien Muchembled's avatar Julien Muchembled
Browse files
Options

CMFActivity: remove broken tests

parent d64e3004
Show whitespace changes
Inline Side-by-side
Showing with 0 additions and 320 deletions
product/CMFActivity/tests/testCMFActivity.py
View file @ 0cf9f566
...@@ -1539,219 +1539,6 @@ class TestCMFActivity(ERP5TypeTestCase, LogInterceptor): ...@@ -1539,219 +1539,6 @@ class TestCMFActivity(ERP5TypeTestCase, LogInterceptor):
message_list = activity_tool.getMessageList() message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list), 2) self.assertEqual(len(message_list), 2)
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)
self.tic()
activity_tool = self.getActivityTool()
def modifySQLAndFail(self, object_list):
# 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],
active_process_uid=[None],
method_id_list=[method_id],
message_list=[pickled_message],
priority_list=[1],
processing_node_list=[-2],
group_method_id_list=['\0'],
tag_list=[''],
order_validation_text_list=[''],
serialization_tag_list=[''],
)
# Mark first entry as failed
del object_list[0][3]
def dummy(self):
pass
try:
Organisation.modifySQLAndFail = modifySQLAndFail
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()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
self.assertEqual(activity_tool.countMessage(method_id='dummy_activity'), 0)
finally:
delattr(Organisation, 'modifySQLAndFail')
delattr(Organisation, 'dummy')
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)
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=['\0'],
tag_list=[''],
order_validation_text_list=[''],
serialization_tag_list=['']
)
# Fail
raise ValueError, 'This method always fail'
try:
Organisation.modifySQLAndFail = modifySQLAndFail
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
obj.activate(activity='SQLQueue').modifySQLAndFail()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
self.assertEqual(activity_tool.countMessage(method_id='dummy_activity'), 0)
finally:
delattr(Organisation, 'modifySQLAndFail')
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)
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=['\0'],
tag_list=[''],
order_validation_text_list=[''],
serialization_tag_list=[''],
)
# Fail
raise ValueError, 'This method always fail'
def dummy(self):
pass
invoke = activity_tool.__class__.invoke
invokeGroup = activity_tool.__class__.invokeGroup
try:
activity_tool.__class__.invoke = modifySQLAndFail
activity_tool.__class__.invokeGroup = modifySQLAndFail
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()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
self.assertEqual(activity_tool.countMessage(method_id='dummy_activity'), 0)
finally:
delattr(Organisation, 'dummy')
activity_tool.__class__.invoke = invoke
activity_tool.__class__.invokeGroup = invokeGroup
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)
self.tic()
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=['\0'],
tag_list=[''],
order_validation_text_list=[''],
serialization_tag_list=[''],
)
# Fail
raise ValueError, 'This method always fail'
def dummy(self):
pass
invoke = activity_tool.__class__.invoke
invokeGroup = activity_tool.__class__.invokeGroup
try:
activity_tool.__class__.invoke = modifySQLAndFail
activity_tool.__class__.invokeGroup = modifySQLAndFail
Organisation.dummy = dummy
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
obj.activate(activity='SQLQueue').dummy()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
self.assertEqual(activity_tool.countMessage(method_id='dummy_activity'), 0)
finally:
delattr(Organisation, 'dummy')
activity_tool.__class__.invoke = invoke
activity_tool.__class__.invokeGroup = invokeGroup
def test_88_ProcessingMultipleMessagesMustRevertIndividualMessagesOnError(self, quiet=0, run=run_all_test): def test_88_ProcessingMultipleMessagesMustRevertIndividualMessagesOnError(self, quiet=0, run=run_all_test):
""" """
Check that, on queues which support it, processing a batch of multiple Check that, on queues which support it, processing a batch of multiple
...@@ -1920,113 +1707,6 @@ class TestCMFActivity(ERP5TypeTestCase, LogInterceptor): ...@@ -1920,113 +1707,6 @@ class TestCMFActivity(ERP5TypeTestCase, LogInterceptor):
LOG('Testing... ',0,message) LOG('Testing... ',0,message)
self.TryUserNotificationRaise('SQLQueue') self.TryUserNotificationRaise('SQLQueue')
def test_94_ActivityToolCommitFailureDoesNotCommitCMFActivitySQLConnectionSQLDict(self, quiet=0, run=run_all_test):
"""
Check that CMFActivity SQL connection is rollback if transaction commit raises.
"""
if not run: return
if not quiet:
message = '\nCheck that activity modifications via CMFActivity connection are rolled back on commit error (SQLDict)'
ZopeTestCase._print(message)
LOG('Testing... ',0,message)
self.tic()
activity_tool = self.getActivityTool()
def modifySQL(self, object_list):
# 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=['\0'],
tag_list=[''],
order_validation_text_list=[''],
)
transaction.get().__class__.commit = fake_commit
commit = transaction.get().__class__.commit
def fake_commit(*args, **kw):
transaction.get().__class__.commit = commit
raise KeyError, 'always fail'
try:
Organisation.modifySQL = modifySQL
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
group_method_id = '%s/modifySQL' % (obj.getPath(), )
obj2 = self.getPortal().organisation_module.newContent(portal_type='Organisation')
self.tic()
obj.activate(activity='SQLDict', group_method_id=group_method_id).modifySQL()
obj2.activate(activity='SQLDict', group_method_id=group_method_id).modifySQL()
self.commit()
try:
self.flushAllActivities(silent=1, loop_size=100)
finally:
transaction.get().__class__.commit = commit
self.assertEqual(activity_tool.countMessage(method_id='dummy_activity'), 0)
finally:
delattr(Organisation, 'modifySQL')
def test_95_ActivityToolCommitFailureDoesNotCommitCMFActivitySQLConnectionSQLQueue(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 commit error (SQLQueue)'
ZopeTestCase._print(message)
LOG('Testing... ',0,message)
self.tic()
activity_tool = self.getActivityTool()
def modifySQL(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=['\0'],
tag_list=[''],
order_validation_text_list=[''],
)
transaction.get().__class__.commit = fake_commit
commit = transaction.get().__class__.commit
def fake_commit(self, *args, **kw):
transaction.get().__class__.commit = commit
raise KeyError, 'always fail'
try:
Organisation.modifySQL = modifySQL
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
self.tic()
obj.activate(activity='SQLQueue').modifySQL()
self.commit()
try:
self.flushAllActivities(silent=1, loop_size=100)
finally:
transaction.get().__class__.commit = commit
self.assertEqual(activity_tool.countMessage(method_id='dummy_activity'), 0)
finally:
delattr(Organisation, 'modifySQL')
def TryActivityRaiseInCommitDoesNotStallActivityConection(self, activity): def TryActivityRaiseInCommitDoesNotStallActivityConection(self, activity):
""" """
Check that an activity which commit raises (as would a regular conflict Check that an activity which commit raises (as would a regular conflict
......
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7