Skip to content

erp5
erp5
Commit 9aca69d4 authored by Vincent Pelletier's avatar Vincent Pelletier
Browse files
Options

Cleanup globally modified properties. 


git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@18505 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent 460c86ab
Hide whitespace changes
Inline Side-by-side
Showing with 71 additions and 47 deletions
product/CMFActivity/tests/testCMFActivity.py
View file @ 9aca69d4
...@@ -1968,18 +1968,22 @@ class TestCMFActivity(ERP5TypeTestCase): ...@@ -1968,18 +1968,22 @@ class TestCMFActivity(ERP5TypeTestCase):
# Remove one entry from object list: this is understood by caller as a # Remove one entry from object list: this is understood by caller as a
# success for this entry. # success for this entry.
object_list.pop() object_list.pop()
Organisation.modifySQLAndFail = modifySQLAndFail
def dummy(self): def dummy(self):
pass pass
Organisation.dummy = dummy try:
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation') Organisation.modifySQLAndFail = modifySQLAndFail
group_method_id = '%s/modifySQLAndFail' % (obj.getPath(), ) Organisation.dummy = dummy
obj.activate(activity='SQLDict', group_method_id=group_method_id).dummy() obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
obj2 = self.getPortal().organisation_module.newContent(portal_type='Organisation') group_method_id = '%s/modifySQLAndFail' % (obj.getPath(), )
obj2.activate(activity='SQLDict', group_method_id=group_method_id).dummy() obj.activate(activity='SQLDict', group_method_id=group_method_id).dummy()
get_transaction().commit() obj2 = self.getPortal().organisation_module.newContent(portal_type='Organisation')
self.flushAllActivities(silent=1, loop_size=100) obj2.activate(activity='SQLDict', group_method_id=group_method_id).dummy()
self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0) get_transaction().commit()
self.flushAllActivities(silent=1, loop_size=100)
self.assertEquals(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): def test_84_ActivityModificationsViaCMFActivityConnectionRolledBackOnErrorSQLQueue(self, quiet=0, run=run_all_test):
""" """
...@@ -2018,12 +2022,15 @@ class TestCMFActivity(ERP5TypeTestCase): ...@@ -2018,12 +2022,15 @@ class TestCMFActivity(ERP5TypeTestCase):
) )
# Fail # Fail
raise ValueError, 'This method always fail' raise ValueError, 'This method always fail'
Organisation.modifySQLAndFail = modifySQLAndFail try:
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation') Organisation.modifySQLAndFail = modifySQLAndFail
obj.activate(activity='SQLQueue').modifySQLAndFail() obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
get_transaction().commit() obj.activate(activity='SQLQueue').modifySQLAndFail()
self.flushAllActivities(silent=1, loop_size=100) get_transaction().commit()
self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0) self.flushAllActivities(silent=1, loop_size=100)
self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0)
finally:
dellatr(Organisation, 'modifySQLAndFail')
def test_85_MessagePathMustBeATuple(self, quiet=0, run=run_all_test): def test_85_MessagePathMustBeATuple(self, quiet=0, run=run_all_test):
""" """
...@@ -2081,19 +2088,26 @@ class TestCMFActivity(ERP5TypeTestCase): ...@@ -2081,19 +2088,26 @@ class TestCMFActivity(ERP5TypeTestCase):
) )
# Fail # Fail
raise ValueError, 'This method always fail' raise ValueError, 'This method always fail'
activity_tool.__class__.invoke = modifySQLAndFail
activity_tool.__class__.invokeGroup = modifySQLAndFail
def dummy(self): def dummy(self):
pass pass
Organisation.dummy = dummy invoke = activity_tool.__class__.invoke
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation') invokeGroup = activity_tool.__class__.invokeGroup
group_method_id = '%s/dummy' % (obj.getPath(), ) try:
obj.activate(activity='SQLDict', group_method_id=group_method_id).dummy() activity_tool.__class__.invoke = modifySQLAndFail
obj2 = self.getPortal().organisation_module.newContent(portal_type='Organisation') activity_tool.__class__.invokeGroup = modifySQLAndFail
obj2.activate(activity='SQLDict', group_method_id=group_method_id).dummy() Organisation.dummy = dummy
get_transaction().commit() obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
self.flushAllActivities(silent=1, loop_size=100) group_method_id = '%s/dummy' % (obj.getPath(), )
self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0) 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)
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): def test_87_ActivityToolInvokeFailureDoesNotCommitCMFActivitySQLConnectionSQLQueue(self, quiet=0, run=run_all_test):
""" """
...@@ -2131,16 +2145,23 @@ class TestCMFActivity(ERP5TypeTestCase): ...@@ -2131,16 +2145,23 @@ class TestCMFActivity(ERP5TypeTestCase):
) )
# Fail # Fail
raise ValueError, 'This method always fail' raise ValueError, 'This method always fail'
activity_tool.__class__.invoke = modifySQLAndFail
activity_tool.__class__.invokeGroup = modifySQLAndFail
def dummy(self): def dummy(self):
pass pass
Organisation.dummy = dummy invoke = activity_tool.__class__.invoke
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation') invokeGroup = activity_tool.__class__.invokeGroup
obj.activate(activity='SQLQueue').dummy() try:
get_transaction().commit() activity_tool.__class__.invoke = modifySQLAndFail
self.flushAllActivities(silent=1, loop_size=100) activity_tool.__class__.invokeGroup = modifySQLAndFail
self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0) 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)
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):
""" """
...@@ -2165,18 +2186,21 @@ class TestCMFActivity(ERP5TypeTestCase): ...@@ -2165,18 +2186,21 @@ class TestCMFActivity(ERP5TypeTestCase):
self.setTitle(self.getTitle() + to_append) self.setTitle(self.getTitle() + to_append)
if fail: if fail:
raise ValueError, 'This method always fail' raise ValueError, 'This method always fail'
Organisation.appendToTitle = appendToTitle try:
obj.setTitle('a') Organisation.appendToTitle = appendToTitle
active_obj.appendToTitle('b') obj.setTitle('a')
active_obj.appendToTitle('c', fail=True) active_obj.appendToTitle('b')
active_obj.appendToTitle('d') active_obj.appendToTitle('c', fail=True)
object_id = obj.getId() active_obj.appendToTitle('d')
get_transaction().commit() object_id = obj.getId()
self.assertEqual(obj.getTitle(), 'a') get_transaction().commit()
self.assertEqual(activity_tool.countMessage(method_id='appendToTitle'), 3) self.assertEqual(obj.getTitle(), 'a')
self.flushAllActivities(silent=1, loop_size=100) self.assertEqual(activity_tool.countMessage(method_id='appendToTitle'), 3)
self.assertEqual(activity_tool.countMessage(method_id='appendToTitle'), 1) self.flushAllActivities(silent=1, loop_size=100)
self.assertEqual(obj.getTitle(), 'abd') self.assertEqual(activity_tool.countMessage(method_id='appendToTitle'), 1)
self.assertEqual(obj.getTitle(), 'abd')
finally:
delattr(Organisation, 'appendToTitle')
def test_suite(): def test_suite():
suite = unittest.TestSuite() 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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7